package com.hexiang.consumer;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;

@Slf4j
@Component
public class MockDeadLetterQueueConsumer {

    private String consumerGroup = "mock-deal-letter-consumer-group";
    private String namesrvAddr = "127.0.0.1:9876";
    private String topicName = "test_topic";

    private DefaultMQPushConsumer consumer;

    @PostConstruct
    public void init() throws MQClientException {
        consumer = consumerInit();
        consumer.start();
    }

    @PreDestroy
    public void shutdown() {
        consumer.shutdown();
    }

    private DefaultMQPushConsumer consumerInit() throws MQClientException {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(consumerGroup);
        consumer.setNamesrvAddr(namesrvAddr);
        consumer.setMessageModel(MessageModel.CLUSTERING);
        // 消费模式:一个新的订阅组第一次启动从队列的最后位置开始消费 后续再启动接着上次消费的进度开始消费
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
        // 订阅主题和 标签（ * 代表所有标签)下信息
        consumer.subscribe(topicName, "*");
        // 设置最大重试次数
        consumer.setMaxReconsumeTimes(1);
        // 注册消费的监听 并在此监听中消费信息, 并返回消费的状态信息
        consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
            // msgs 中只收集同一个topic, 同一个 tag, 并且 key 相同的message
            // 会把不同的消息分别放置到不同的队列中
            for (MessageExt messageExt : msgs) {
                try {
                    //逐条消费
                    String messageBody = new String(messageExt.getBody());
                    log.info("Message Consumer: Handle New Message: messageId: " + messageExt.getMsgId() + ",topic: " +
                            messageExt.getTopic() + ",tags: " + messageExt.getTags() + ",messageBody: " + messageBody);

                    //模拟业务异常
                    int i = 1 / 0;
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                } catch (Exception e) {
                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                }
            }
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        });
        return consumer;
    }
}
